package com.kafka.dongli.config.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/***
 * 自定义生产者发送消息拦截器
 */
public class CustomerProducerInterceptor implements ProducerInterceptor<String, Object> {


    // 存储已处理的消息标识（key：消息唯一标识，value：处理时间戳）
    private final ConcurrentHashMap<String, Long> processedMessages = new ConcurrentHashMap<>();
    // 清理过期记录的阈值（如1小时前的记录）
    private static final long CLEAN_THRESHOLD_MS = TimeUnit.HOURS.toMillis(1);

    /**
     * 发送消息时，会先调用该方法，对消息进行拦截，可以在拦截中对消息做一些处理，记录日志等等操作.....
     *
     * @param record the record from client or the record returned by the previous interceptor in the chain of interceptors.
     * @return
     */
    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord record) {

        // 发送前的处理（如添加唯一标识到消息头）
        // 这里假设消息key唯一，或使用record.headers()添加自定义唯一ID

        System.out.println("拦截消息：" + record.toString());
        return record;
    }

    /**
     * 生产者 发送消息到服务器 收到消息后的一个确认
     *
     * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
     *                 If an error occurred, metadata will contain only valid topic and maybe
     *                 partition. If partition is not given in ProducerRecord and an error occurs
     *                 before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
     *                 The metadata may be null if the client passed null record to
     *                 {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {


        if (metadata != null) {
            System.out.println("服务器收到该消息：" + metadata.offset());
        } else {
            System.out.println("消息发送失败了，exception = " + exception.getMessage());
        }

//        if (metadata == null) {
//            // 消息完全发送失败（无元数据），可根据需要处理
//            return;
//        }
//
//        // 生成消息唯一标识：topic + partition + offset（确保全局唯一）
//        String messageId = metadata.topic() + "-" + metadata.partition() + "-" + metadata.offset();
//
//        // 检查是否已处理过
//        if (processedMessages.containsKey(messageId)) {
//            // 已处理，直接返回
//            return;
//        }
//
//        // 执行实际业务逻辑（如统计、日志等）
//        try {
//            if (exception == null) {
//                System.out.println("消息发送成功，执行业务逻辑：" + messageId);
//                // TODO: 你的业务逻辑（如更新数据库、计数等）
//            } else {
//                System.out.println("消息发送失败，执行失败逻辑：" + messageId + "，异常：" + exception.getMessage());
//                // TODO: 失败处理逻辑
//            }
//        } finally {
//            // 标记为已处理
//            processedMessages.put(messageId, System.currentTimeMillis());
//            // 定期清理过期记录，避免内存泄漏
//            cleanExpiredRecords();
//        }

    }

    // 清理过期的处理记录（避免内存无限增长）
    private void cleanExpiredRecords() {
        long now = System.currentTimeMillis();
        processedMessages.entrySet().removeIf(
                entry -> now - entry.getValue() > CLEAN_THRESHOLD_MS
        );
    }

    @Override
    public void close() {
        // 资源清理
        processedMessages.clear();
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
